package com.sh.data.engine.domain.datadev.offline.runner;

import cn.hutool.extra.spring.SpringUtil;
import com.alibaba.fastjson.JSONArray;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.google.common.collect.Lists;
import com.sh.data.engine.common.constants.HttpStatus;
import com.sh.data.engine.common.constants.code.BizCodes;
import com.sh.data.engine.common.enumDefinition.DSType;
import com.sh.data.engine.common.enumDefinition.DataQualityRuleCompareTypeEnum;
import com.sh.data.engine.common.exception.BusinessException;
import com.sh.data.engine.common.util.TableNameUtil;
import com.sh.data.engine.domain.common.service.CommonService;
import com.sh.data.engine.domain.datadev.offline.model.enums.OfflineDevJobStatusEnum;
import com.sh.data.engine.domain.datadev.offline.service.DatadevDataqualityService;
import com.sh.data.engine.domain.dataquality.model.domain.DataQualityRuleTemplateDomain;
import com.sh.data.engine.domain.dataquality.service.DataQualityManagerService;
import com.sh.data.engine.domain.dataquality.utils.DataQualityUtils;
import com.sh.data.engine.domain.integration.datasource.service.DataSourceService;
import com.sh.data.engine.domain.shims.db.BaseDbManager;
import com.sh.data.engine.domain.shims.db.model.FieldInfoDomain;
import com.sh.data.engine.domain.shims.sql.builder.QueryBuilder;
import com.sh.data.engine.domain.util.ConvertUtil;
import com.sh.data.engine.domain.util.VarParserUtil;
import com.sh.data.engine.repository.dao.datadev.entity.DatadevDataqualityEntity;
import com.sh.data.engine.repository.dao.datadev.entity.DatadevJobEntity;
import com.sh.data.engine.repository.dao.datadev.entity.DatadevMenuNodeEntity;
import com.sh.data.engine.repository.dao.dataquality.entity.DataQualityRuleEntity;
import com.sh.data.engine.repository.dao.workspace.manager.entity.StorageEntity;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.hive.jdbc.HiveStatement;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

import static com.sh.data.engine.domain.dataquality.utils.DataQualityUtils.*;

/**
 * @author: mengzheng.mhc
 * @date: 2025/9/11 11:39
 */
@Slf4j
public class OfflineDevDataQualityJobRunner extends OfflineDevBaseJobRunner {
    private DataSourceService dataSourceService = SpringUtil.getBean(DataSourceService.class);
    private DataQualityManagerService dataQualityManagerService =
        SpringUtil.getBean(DataQualityManagerService.class);
    private CommonService commonService = SpringUtil.getBean(CommonService.class);
    private DatadevDataqualityService datadevDataqualityService =
        SpringUtil.getBean(DatadevDataqualityService.class);

    public OfflineDevDataQualityJobRunner(
        DatadevJobEntity jobEntity,
        DatadevMenuNodeEntity menuNodeEntity,
        Map<Long, Integer> parentNodeIdAndLineStatusMap,
        List<Long> childNodeIdList,
        Map<Long, OfflineDevBaseJobRunner> allJobNodeMap,
        ExecutorService executorService) {
        super(
            jobEntity,
            menuNodeEntity,
            parentNodeIdAndLineStatusMap,
            childNodeIdList,
            allJobNodeMap,
            executorService);
    }

    @Override
    public void cancel() {
        // todo
    }

    @Override
    public int runJob() throws Exception {

        DatadevDataqualityEntity dataqualityTaskConfig =
            datadevDataqualityService.getOne(
                new LambdaQueryWrapper<DatadevDataqualityEntity>()
                    .eq(DatadevDataqualityEntity::getNodeId, menuNodeEntity.getId()));

        Long projectId = menuNodeEntity.getProjectId();
        StorageEntity storageEntity = commonService.getStorage(jobEntity.getProjectId());
        String storageType = storageEntity.getStorageType();

        // 1.查询表和数据源连接
        String tblName = dataqualityTaskConfig.getTblName();

        List<FieldInfoDomain> fieldListProjectId =
            dataSourceService.getFieldListProjectId(tblName, projectId);
        if (CollectionUtils.isEmpty(fieldListProjectId)) {
            throw new BusinessException("数据表不存在");
        }

        // 1.1 如果是pg和sap，要对表名加上引号
        if (storageType.equals(DSType.PostgreSQL.name())
            || storageType.equals(DSType.MatrixDB.name())
            || storageType.equals(DSType.Sap.name()) || storageType.equals(DSType.KINGBASE8.name())) {
            tblName = TableNameUtil.getTblNameLong(tblName);
        } else if (storageType.equals(DSType.DM.name())) {
            String dbName = storageEntity.getRmdbs().getDbName();
            tblName = "\"" + dbName + "\"." + "\"" + tblName + "\"";
        }

        DataQualityRuleTemplateDomain ruleTemplateByTemplate = new DataQualityRuleTemplateDomain();
        String diySql = "";

        // 2.查询模版
        if (dataqualityTaskConfig.getRuleDiy() == null
            || dataqualityTaskConfig.getRuleDiy().equals(0)) {
            ruleTemplateByTemplate =
                dataQualityManagerService.getRuleTemplateByTemplateId(
                    dataqualityTaskConfig.getTemplateId());
            if (ruleTemplateByTemplate == null) {
                throw new BusinessException("查询不到规则模版");
            }

        } else {
            // 如果是自定义模板
            ruleTemplateByTemplate = null;
            diySql = dataqualityTaskConfig.getDiySql();
            //      diySql =
            //          String.format(
            //              "select %s(${field}) from %s",
            //              dataqualityTaskConfig.getRuleDiyStatisticalModel(), tblName);
        }
        // 3.查询分区 现在的工作流这里前端是不会传partitionExpression过来的，partitionExpression为null
        String partitionExpression = dataqualityTaskConfig.getPartitionExpression();
        try {
            if (StringUtils.isNotBlank(partitionExpression)) {
                partitionExpression = VarParserUtil.getDataqualityPt(partitionExpression);
            }
        } catch (Exception e) {
            throw new BusinessException("分区表达式格式错误");
        }

        // 4.执行任务
        String sqlTemp;

        String[] fields = {"-"};
        String[] fieldTypes = {"-"};

        if (dataqualityTaskConfig.getRuleDiy().equals(1)) {
            // 自定义规则
            StringBuilder sb = new StringBuilder();
            sb.append(diySql);
            if (StringUtils.isNotBlank(partitionExpression)) {
                sb.append(" ").append("where").append(" ").append(partitionExpression);
            }
            sqlTemp = sb.toString();
        } else {
            // 根据不同的存储拿到对应的模版sql
            sqlTemp =
                DataQualityUtils.replaceTableParam(
                    ruleTemplateByTemplate, partitionExpression, tblName, storageType);

            fields = dataqualityTaskConfig.getFieldNames().split(",");
            fieldTypes = dataqualityTaskConfig.getFieldTypes().split(",");
        }

        // sql都没有直接报错
        if (StringUtils.isBlank(sqlTemp)) {
            this.writeLog(Lists.newArrayList("sql错误"));
            return OfflineDevJobStatusEnum.FAILURE.getCode();
        }

        boolean taskException = false;

        // 挨个执行字段
        for (int i = 0; i < fields.length; i++) {
            String execSql =
                DataQualityUtils.getExecSql(
                    fields[i],
                    fieldTypes[i],
                    sqlTemp,
                    dataqualityTaskConfig.getRuleCompareType(),
                    dataqualityTaskConfig.getRuleCondition(),
                    ruleTemplateByTemplate,
                    storageType,
                    dataqualityTaskConfig.getTemplateFilter());

            try {

                String querySql = "\n> execute sql : " + execSql;
                writeLog(Lists.newArrayList(querySql));

                // 执行sql并获取执行后值
                String value = doSql(execSql, jobEntity.getProjectId(), storageType);

                if (StringUtils.isNotBlank(value)) {
                    String log = String.format("sql:%s 执行结果:%s", execSql, value);
                    writeLog(Lists.newArrayList(log));
                }

                // 根据规则检查
                DataQualityRuleEntity dataQualityRuleEntity =
                    ConvertUtil.copyProperties(dataqualityTaskConfig, DataQualityRuleEntity.class);
                Boolean flag = DataQualityUtils.checkRule(dataQualityRuleEntity, value);
                if (!flag) {
                    throw new BusinessException("异常");
                }

                String log = String.format("字段:%s校验通过", fields[i]);
                writeLog(Lists.newArrayList(log));

            } catch (Exception e) {
                String log = "";
                if (e instanceof BusinessException) {
                    BusinessException businessException = (BusinessException) e;
                    log =
                        this.formatWarnContent(
                            fields[i],
                            businessException.getCode(),
                            dataqualityTaskConfig,
                            ruleTemplateByTemplate);
                } else {
                    log = "系统异常," + ExceptionUtils.getFullStackTrace(e);
                }

                writeLog(Lists.newArrayList(log));

                taskException = true;
            }
            if (taskException) {
                return OfflineDevJobStatusEnum.FAILURE.getCode();
            }
        }

        return OfflineDevJobStatusEnum.SUCCESS.getCode();
    }

    private String formatWarnContent(
        String fieldName,
        Integer status,
        DatadevDataqualityEntity rule,
        DataQualityRuleTemplateDomain template) {
        StringBuilder content = new StringBuilder("字段:");
        content.append(fieldName).append("出现:");
        if (status == 3) {
            content.append("sql语句出错");
            return content.toString();
        }
        content.append("数据质量问题;");

        String compareType = "";
        if (StringUtils.isNotBlank(rule.getRuleCompareType())
            && !"null".equals(rule.getRuleCompareType())) {
            if (rule.getRuleCompareType().split(",").length == 1) {
                if (null
                    != DataQualityRuleCompareTypeEnum.getByCompareType(
                    Integer.valueOf(rule.getRuleCompareType()))) {
                    compareType =
                        DataQualityRuleCompareTypeEnum.getByCompareType(
                                Integer.valueOf(rule.getRuleCompareType()))
                            .getDesc();
                }
            }
        }

        String msg = "";
        String min;
        String max;
        if (rule.getRuleDiy() == 0) {
            switch (String.valueOf(template.getId())) {
                case WZXYS:
                    msg = String.format("唯一值记录数和总记录数的比值%s%s", compareType, rule.getRuleExpectValue());
                    break;
                case KZYS:
                    msg = "存在空值";
                    break;
                case KZZBYS:
                    msg = String.format("空值占比%s%s", compareType, rule.getRuleExpectValue());
                    break;
                case QZFWYS:
                    //          String min = this.getCondition(rule.getRuleCondition(), "min");
                    //          String max = this.getCondition(rule.getRuleCondition(), "max");

                    msg = String.format("取值范围约束不匹配");
                    break;
                case ZDZYS:
                    max = this.getCondition(rule.getRuleCondition(), "max");
                    msg = String.format("最大值大于%s", max);
                    break;
                case ZXZYS:
                    min = this.getCondition(rule.getRuleCondition(), "min");
                    msg = String.format("最小值小于%s", min);
                    break;
                case PJZYS:
                    String avg = this.getCondition(rule.getRuleCondition(), "avg");
                    msg = String.format("平均值%s%s", compareType, avg);
                    break;
                case ZZYS:
                    String sum = this.getCondition(rule.getRuleCondition(), "sum");
                    msg = String.format("总值%s%s", compareType, sum);
                    break;
                case CDYS:
                    // String length = this.getCondition(rule.getRuleCondition(), "length");
                    msg = String.format("长度约束规则不匹配");
                    break;
                case DMCDYS:
                    msg = String.format("非指定长度的记录数%s%s", compareType, rule.getRuleExpectValue());
                    break;
                case ZZBDSYS:
                    msg = "存在非身份证号格式的值";
                    break;
                case RQZYS:
                    String date = this.getCondition(rule.getRuleCondition(), "date");
                    msg = String.format("日期值%s%s", compareType, date);
                    break;
                case MJZYS:
                    String enumVal = this.getCondition(rule.getRuleCondition(), "enum");
                    msg = String.format("枚举值%s[%s]", compareType, enumVal);
                    break;
                case KZFCYS:
                    msg = "值存在空字符串";
                    break;
                case KZHZFCYS:
                    msg = "值存在空值或空字符串";
                    break;
                case TSZFYS:
                    String specialChar = this.getCondition(rule.getRuleCondition(), "specialChar");
                    msg = String.format("值存在特殊字符%s", specialChar);
                    break;
                case CFZYS:
                    msg = "存在重复值";
                    break;
                case JSXYST:
                    msg = "数据在天级别存在延迟";
                    break;
                case ZZBDSYSDH:
                    msg = "存在非电话号码格式的值";
                    break;
                case JSXYSX:
                    msg = "数据在小时级别存在延迟";
                    break;
                default:
            }
        } else {
            msg =
                String.format(
                    "%s值%s%s", rule.getRuleDiyStatisticalModel(), compareType, rule.getRuleExpectValue());
        }
        content.append(msg);
        return content.toString();
    }

    private String doSql(String execSql, Long projectId, String storageType) {
        CountDownLatch latch = new CountDownLatch(1);
        Function<List<String>, Void> writeLog =
            (List<String> logs) -> {
                this.writeLog(logs);
                return null;
            };

        BaseDbManager dbManager = commonService.getStorageDbManagerByProjctId(projectId);

        QueryBuilder queryBuilder = null;
        ResultSet rs = null;

        try {
            if (storageType.equals(DSType.Hive.name())) {
                HiveStatement stmt = (HiveStatement) dbManager.getConnection().createStatement();
                processHiveLog(stmt, latch, writeLog);
                boolean hasRS = stmt.execute(execSql);
                if (hasRS) {
                    rs = stmt.getResultSet();
                }
            } else {
                // 除了hive的其他数据源 pg sap tidb
                String time = DateFormatUtils.format(new Date(), "yyyy-MM-dd HH:mm:ss.SSS");
                String format = String.format("[%s] 当前数据源:%s", time, storageType);
                List<String> logInfo = Lists.newArrayList(format);

                long l = System.currentTimeMillis();
                queryBuilder = dbManager.getQueryBuilder();
                rs = queryBuilder.selectSql(execSql).list();

                String queryStatus = "> execute status: ok ";
                long l2 = System.currentTimeMillis();
                String queryTime = "> query time: " + (l2 - l) / 1000.0d;
                logInfo.add(queryStatus);
                logInfo.add(queryTime);
                writeLog.apply(logInfo);
            }

            String value = null;
            if (rs.next()) {
                value = rs.getString(1);
            }
            rs.close();
            return value;

        } catch (Exception e) {
            log.error(e.getMessage(), e);
            List<String> errorInfo = Lists.newArrayList(e.toString());
            writeLog.apply(errorInfo);
            throw new BusinessException(BizCodes.DATA_QUALITY_SQL_STATEMENT_ERROR, HttpStatus.INTERNAL_SERVER_ERROR);
        } finally {
            if (rs != null) {
                try {
                    rs.close();
                } catch (SQLException e) {
                    log.error(e.getMessage(), e);
                }
            }
            if (storageType.equals(DSType.Hive.name())) {
                await(latch);
            } else {
                queryBuilder.close();
            }
        }
    }

    private String doHiveSql(String execSql, Connection conn) {

        CountDownLatch latch = new CountDownLatch(1);
        Function<List<String>, Void> writeLog =
            (List<String> logs) -> {
                this.writeLog(logs);
                return null;
            };
        try {
            HiveStatement stmt = (HiveStatement) conn.createStatement();
            processHiveLog(stmt, latch, writeLog);
            boolean hasRS = stmt.execute(execSql);
            if (hasRS) {
                ResultSet rs = stmt.getResultSet();
                String value = null;
                if (rs.next()) {
                    value = rs.getString(1);
                }
                rs.close();
                return value;
            }
            return null;
        } catch (Exception e) {
            log.error(e.getMessage(), e);
            throw new BusinessException("sql语句出错");
        } finally {
            await(latch);
        }
    }

    private void await(CountDownLatch latch) {
        try {
            latch.await();
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private void processHiveLog(
        HiveStatement stmt, CountDownLatch latch, Function<List<String>, Void> writeLogFunction) {
        new Thread(
            () -> {
                try {
                    while (!stmt.isClosed() && stmt.hasMoreLogs()) {
                        List<String> logList = stmt.getQueryLog();
                        if (CollectionUtils.isEmpty(logList)) {
                            try {
                                TimeUnit.MICROSECONDS.sleep(100);
                            } catch (Exception ignore) {
                            }
                            continue;
                        }

                        writeLogFunction.apply(logList);
                    }
                } catch (Exception e) {
                    log.error("日志读取失败{}", e.getMessage());
                } finally {
                    latch.countDown();
                }
            })
            .start();
    }

    private String getCondition(String ruleCondition, String key) {
        try {
            if (StringUtils.isNotBlank(ruleCondition)) {
                List<RuleCondition> ruleConditions =
                    JSONArray.parseArray(ruleCondition, RuleCondition.class);

                if (org.apache.commons.collections.CollectionUtils.isNotEmpty(ruleConditions)) {
                    Optional<RuleCondition> optional =
                        ruleConditions.stream()
                            .filter(c -> StringUtils.equalsIgnoreCase(c.getKey(), key))
                            .findFirst();
                    if (optional.isPresent()) {
                        return optional.get().getValue();
                    }
                }
            }
        } catch (Exception ignore) {
        }
        return null;
    }

    @Data
    private static class RuleCondition {

        private String key;

        private String value;
    }
}
